﻿using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.Linq;
using System.Text;
using System.Threading;
using Common.Utilities;
using Common.Utilities.Dispatcher;

namespace Workflows.Components.TextMining.Collapse
{
    /// <summary>
    /// 
    /// </summary>
    public class MergeCollapseWorker:IWorker 
    {
        #region Implementation of IWorker

        public event WorkerStarted WorkerStartedEvent;
        public event WorkerFinished WorkerFinishedEvent;
        public event WorkerHasError WorkerErredEvent;
        public event WorkerCancelled WorkerCancelledEvent;
        public event WorkerStatusChanged WorkerStatusChangedEvent;

        private Thread workerThread;
        private MergeCollapseInstruction instruction;
        private MergeCollapseOutput output;
        private RunContext _Context;
        private Guid _TicketID;
        private string _WorkflowName;
        private string _ComponentName;

        /// <summary>
        /// 
        /// </summary>
        public string ComponentName
        {
            get { return _ComponentName; }
        }

        /// <summary>
        /// 
        /// </summary>
        public string WorkflowName
        {
            get { return _WorkflowName; }
        }

        /// <summary></summary>
        public Thread WorkerThread
        {
            get { return workerThread; }
        }
        /// <summary></summary>
        public InstructionBase Instruction
        {
            get { return instruction; }
        }
        /// <summary></summary>
        public ResultBase Output
        {
            get { return output; }
        }
        /// <summary></summary>
        public Type ComponentType
        {
            get { return typeof(MergeCollapseComponent); }
        }
        /// <summary></summary>
        public RunContext Context
        {
            get { return _Context; }
            set { _Context = value; }
        }
        /// <summary></summary>
        public void SetInstruction(Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._TicketID = ticketGuid;
            this.instruction = (MergeCollapseInstruction)request;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }
        /// <summary></summary>
        public void SetInstruction(RunContext context, Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._TicketID = ticketGuid;
            this.instruction = (MergeCollapseInstruction)request;
            this._Context = context;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }
        /// <summary></summary>
        public void DoWork()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._TicketID, this, this.GetType(), this.instruction);
            }
            this.workerThread = Thread.CurrentThread;
            this.MergeCollapsedEntities();
        }
        /// <summary></summary>
        public void DoWorkAsync()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._TicketID, this, this.GetType(), this.instruction);
            }
            this.workerThread = new Thread(new ThreadStart(this.MergeCollapsedEntities));
            this.workerThread.Start();
        }
        /// <summary></summary>
        public void CancelWork()
        {
            if (this.workerThread != null && this.workerThread.IsAlive)
            {
                this.workerThread.Abort();
                this.workerThread.Join();

                if (this.WorkerCancelledEvent != null)
                {
                    this.WorkerCancelledEvent(this._TicketID, this.GetType(), this.instruction);
                }
            }
        }

        #endregion

        #region Work
        private void MergeCollapsedEntities()
        {
            SqlConnection conn=new SqlConnection(this.instruction.ConnStr);
            try
            {
                conn.Open();
                SqlCommand cmd = conn.CreateCommand();
                cmd.CommandType = CommandType.StoredProcedure;
                int totalEntities = 0;

                string sp_Name = string.Format("bulkinsert_{0}", this.instruction.MergeTableName);
                TableColumn[] srcCols1 = DbUtil.GetTableColumns(
                    DataProviderType.MSSQL, this.instruction.ConnStr,
                    this.instruction.SchemaName,
                    this.instruction.CollapseMappingTablePrefix + this.instruction.PartFrom.ToString());
                TableColumn[] srcCols2 = DbUtil.GetTableColumns(
                    DataProviderType.MSSQL, this.instruction.ConnStr,
                    this.instruction.SchemaName,
                    this.instruction.CollapseOutputTablePrefix + this.instruction.PartFrom.ToString());
                List<TableColumn> cols=new List<TableColumn>();
                Dictionary<string,SqlParameter> sqlParameters=new Dictionary<string, SqlParameter>();
                foreach(TableColumn col1 in srcCols1)
                {
                    cols.Add(col1);
                    sqlParameters.Add(col1.ColumnName.ToLower(),
                        new SqlParameter("@"+col1.ColumnName,DBNull.Value));
                }
                foreach(TableColumn col2 in srcCols2)
                {
                    if(!sqlParameters.ContainsKey(col2.ColumnName.ToLower()))
                    {
                        cols.Add(col2);
                        sqlParameters.Add(col2.ColumnName.ToLower(),
                            new SqlParameter("@"+col2.ColumnName,DBNull.Value));
                    }
                }
                TableColumn profileCol=new TableColumn();
                profileCol.ColumnName = this.instruction.MergeProfileFieldName;
                profileCol.DbType = SqlDbType.Int;
                cols.Add(profileCol);
                sqlParameters.Add(this.instruction.MergeProfileFieldName.ToLower(),
                    new SqlParameter("@"+this.instruction.MergeProfileFieldName, DBNull.Value));

                DbUtil.NewTable(DataProviderType.MSSQL,
                                this.instruction.ConnStr, this.instruction.SchemaName,
                                this.instruction.MergeTableName, cols.ToArray());
                string sp_Body = StoredProcUtil.CreateStoredProcedureStatementForInsert(
                    DataProviderType.MSSQL, sp_Name, this.instruction.SchemaName,
                    this.instruction.MergeTableName, cols.ToArray());
                DbUtil.NewStoredProc(this.instruction.ConnStr, sp_Name, sp_Body);
                cmd.CommandText = sp_Name;

                int profileID = 0;
                for(int i=this.instruction.PartFrom; i<=this.instruction.PartTo;i++)
                {
                    string sql = string.Format(
                        "select [{0}],[{1}],[{2}] from {3}.[{4}]",
                        this.instruction.CollapseIDField,
                        this.instruction.IDField,
                        this.instruction.ScoreField,
                        this.instruction.SchemaName,
                        this.instruction.CollapseMappingTablePrefix + i.ToString());
                    DataTable dt1 = DbUtil.GetTableDataBySQL(
                        this.instruction.ConnStr, "Table1", sql);
                    int maxCollapseID = 0;
                    foreach(DataRow dr in dt1.Rows)
                    {
                        if (this.WorkerStatusChangedEvent != null && maxCollapseID % 10 == 0)
                        {
                            string msg = string.Format(
                                "Merging part {0} of {1}, {2} of {3} records collapsed",
                                i-this.instruction.PartFrom,
                                this.instruction.PartTo - this.instruction.PartFrom,
                                maxCollapseID+ profileID, totalEntities);
                            int pct = (i - this.instruction.PartFrom)*100/
                                      (this.instruction.PartTo - this.instruction.PartFrom + 1);
                            this.WorkerStatusChangedEvent(this._TicketID, this.GetType(), this.instruction,
                                                          this._WorkflowName, this._ComponentName, msg, pct);
                        }

                        int collapseID = int.Parse(dr[this.instruction.CollapseIDField].ToString());
                        if(collapseID>maxCollapseID)
                            maxCollapseID = collapseID;
                        int entityProfileID = profileID + collapseID;
                        int id = int.Parse(dr[this.instruction.IDField].ToString());
                        decimal score = 0;
                        try
                        {
                            score = decimal.Parse(dr[this.instruction.ScoreField].ToString());
                        }
                        catch
                        {
                            score = 0;
                        }
                        string sql2 = string.Format(
                            "select * from {0}.[{1}] where [{2}]={3}",
                            this.instruction.SchemaName,
                            this.instruction.CollapseOutputTablePrefix + i.ToString(),
                            this.instruction.CollapseIDField,
                            collapseID);

                        DataTable dt2 = DbUtil.GetTableDataBySQL(
                            this.instruction.ConnStr, "Table2", sql2);
                        foreach(DataRow dr2 in dt2.Rows)
                        {
                            totalEntities++;
                            foreach(DataColumn col in dt2.Columns)
                            {
                                if (sqlParameters.ContainsKey(col.ColumnName.ToLower()))
                                    sqlParameters[col.ColumnName.ToLower()].Value = dr2[col.ColumnName];
                            }
                            sqlParameters[this.instruction.MergeProfileFieldName.ToLower()].Value = entityProfileID;
                            sqlParameters[this.instruction.IDField.ToLower()].Value = id;
                            sqlParameters[this.instruction.ScoreField.ToLower()].Value = score;
                            cmd.Parameters.Clear();
                            foreach (string key in sqlParameters.Keys)
                                cmd.Parameters.Add(sqlParameters[key]);
                            cmd.ExecuteNonQuery();
                        }
                    }
                    profileID += maxCollapseID;
                }

                this.output = new MergeCollapseOutput(
                    this.instruction.OwnerComponentID, this._TicketID,
                    this.instruction.MergeTableName, profileID, totalEntities);
                if (this.WorkerFinishedEvent != null)
                    this.WorkerFinishedEvent(this._TicketID, this.GetType(), this.instruction, this.output);
            }
            catch(Exception ex)
            {
                if (this.WorkerErredEvent != null)
                    this.WorkerErredEvent(this._TicketID, this.GetType(), this.instruction, ex.ToString(), true);
            }
        }
        #endregion
    }
}
